<?php
namespace app\helpclass;


use Kafka\Producer;
use Kafka\ProducerConfig;
use RuntimeException;
use Yii;
use Zipkin\Recording\Span;
use Zipkin\Reporter;


class KafkaReporter implements Reporter
{
    public $broker_list = "192.168.99.100:9092";
    public $topic = "sleuth";

    public function __construct()
    {
        $this->broker_list = Yii::$app->params['kafka_host'].":".Yii::$app->params['kafka_port'];
        $this->topic = Yii::$app->params['kafka_topic'];
    }

    /**
     * 将zipkin数据报告到kafka里
     * @param array $spans
     */
    public function report(array $spans)
    {
        try{
            $payload = json_encode(array_map(function (Span $span) {
                return $span->toArray();
            }, $spans));
            //Yii::info('payload:'.$payload,'myinfo');
            $config = ProducerConfig::getInstance();
            $config->setMetadataRefreshIntervalMs(10000);
            $config->setMetadataBrokerList($this->broker_list);
            //$config->setMetadataBrokerList('192.168.99.100:9092');
            //$config->setMetadataBrokerList('10.35.165.16:9092');测试环境地址
            $config->setRequiredAck(1);
            $config->setIsAsyn(false);
            $config->setProduceInterval(500);

            $producer = new Producer();
            $producer->send(  [
                [
                    'topic' => $this->topic,
                    'value' => $payload,
                ]
            ]);
        }catch (RuntimeException $exception)
        {
            Yii::info(__METHOD__.$exception->getMessage());
        }
    }
}